package com.qilin.hscs.hm.mqtt.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.qilin.hscs.hm.mqtt.constant.HostNameConstant;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * Created by 10188 on 2018/3/23.
 */
@Component
@Order(4)
public class InnerHostRepeater extends Repeater implements HostNameConstant {

    private static final Logger logger= LogManager.getLogger(InnerHostRepeater.class.getName());

    private MqttClient clienta;

    private MqttClient clientg;

    private MqttClient repeaterClient;

    private String clientIda = "innerClienta";

    private String clientIdg = "innerClientg";

    private String repeaterClientId="innerRepeaterClient";

    JSONObject msgObj;
    @Autowired
    @Qualifier("innerHost")
    private String innerHost;

    @Override
    public void repeat() throws MqttException {
        msgObj = JSON.parseObject(msg);
        msgObj.put("isSelf", true);
        msg = JSON.toJSONString(msgObj);
        repeaterClient=factory.getClient(repeaterClientId,innerHost);
        casUtil.send(repeaterClient,msg,topicVar);
    }

    @Override
    @PostConstruct
    void listenFromApp() throws Exception {
        logger.info("innerClient启动");
        clienta = factory.getClient(clientIda, innerHost);
        casUtil.receive(clienta, topicFromApp, new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {

            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                msg = new String(mqttMessage.getPayload());
                try {
                    msgObj = JSON.parseObject(msg);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                Boolean isSelf = msgObj.getBoolean("isSelf");
                if (isSelf==null) {
                    logger.info("【收到内app发布的消息】={}",msg);
                    topicVar=s;
                    flushToHandler();
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        });
    }

    @Override
    void flushToHandler() {
        Observable<String> observable= Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext(innerHostName+"#"+topicVar+"#"+msg);
            }
        });

        observable.subscribe(handler);
    }

    @PostConstruct
    void listenFromGateway() throws Exception {
        logger.info("监听网关的客户端已启动");
        clientg = factory.getClient(clientIdg, innerHost);
        casUtil.receive(clientg, topicFromGateway, new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {

            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                msg = new String(mqttMessage.getPayload());
                msgObj = JSON.parseObject(msg);
                Boolean isSelf = msgObj.getBoolean("isSelf");
                if (isSelf==null) {
                    logger.info("【收到网关发布的消息】={}",msg);
                    topicVar=s;
                    flushToHandler();
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        });
    }

    @Override
    public void onSubscribe(Disposable disposable) {

    }

    @Override
    public void onNext(String s) {
        logger.info("【收到处理中心发送的消息】={}",s);
        String[] content = s.split("#");
        String hostName = content[0];
        topicVar = content[1];
        msg = content[2];
        logger.debug(hostName + " " + topicVar + " " + msg);
        try {
            repeat();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
}
